package demo;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 *1、算子之间的传输关系：一对一，重分区
 *
 * 2、算子串在一起的条件
 * 1）一对一
 * 2）并行度相同
 *
 * 3、关于算子链的API
 * 1）全局禁用
 * 2）某个算子不参与链化，A。disableChaining() 算子A不会和前面和后面的算子串在一起
 * 3）某个算子 ，A.startNewChain()，算子A不与前面串在一起，从A开始正常链化
 */
public class Flink05_算子链 {


    public static void main(String[] args) throws Exception {

        /**
         * 创建执行环境
         * IDEA运行的时候，
         * 需要导入依赖
         */
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        /**
         * 设置全局并行度，全局并行度的优先级低于在算子上指定，默认的并行度和线程数有关，Socket这种只有一个并行度。
         * 也可以在WEB界面，提交JAR包的时候设置并行度。
         */
        executionEnvironment.setParallelism(3);


        /**
         * 全局禁用算子链
         */
        executionEnvironment.disableOperatorChaining();

        /**
         * 读取数据，此处为从文件中读取数据
         */

        DataStreamSource<String> stringDataStreamSource = executionEnvironment.socketTextStream("192.168.159.104",7777);

        /**
         * 进行切分，分组，聚合
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] s1 = s.split(" ");

                for (String s2 : s1) {
                    Tuple2<String, Integer> tuple2 = new Tuple2<>(s2, 1);
                    collector.collect(tuple2);
                }
            }
        });

        /**
         * Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成，
         * 根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能
         *
         * 如果用一下的写法，会报错，因为Flink不知道返回的Tuple中的字段的具体类型是什么,需要进行显式的指定
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> lamda = stringDataStreamSource.flatMap(
                (String s, Collector<Tuple2<String, Integer>> collector)->{
                    String[] s1 = s.split(" ");
                    for (String s2 : s1) {
                        Tuple2<String, Integer> tuple2 = new Tuple2<>(s2, 1);
                        collector.collect(tuple2);
                    }
                }
        ).returns(Types.TUPLE(Types.STRING,Types.INT));


        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = streamOperator.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) {
                return stringIntegerTuple2.f0;
            }
        });



        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum(1);


        sum.print();

        executionEnvironment.execute();


    }
}
